package com.luxf.base.domainevent;

import com.luxf.base.domainevent.codec.MessageDecoder;
import com.luxf.base.domainevent.context.LocalUser;
import com.luxf.base.domainevent.context.LocalUserHolder;
import com.luxf.base.domainevent.eventtype.EventTypeLoader;
import com.luxf.base.domainevent.model.DomainEventNotice;
import com.luxf.base.domainevent.model.MessageReceivedHistory;
import com.luxf.base.domainevent.repository.MessageReceivedHistoryRepository;
import com.luxf.base.domainevent.utils.EventNoticeUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;

import javax.annotation.Resource;
import java.time.Instant;

/**
 * @author fast66
 * 2021-05-26 15:57
 **/
@Slf4j
public class DefaultDomainEventNoticeHandler implements DomainEventNoticeHandler {

    @Resource
    private PlatformTransactionManager platformTransactionManager;

    @Resource
    private TransactionDefinition transactionDefinition;

    @Resource
    private MessageReceivedHistoryRepository messageReceivedHistoryRepository;

    @Resource
    private ApplicationEventPublisher applicationEventPublisher;

    @Resource
    private EventTypeLoader eventTypeLoader;

    @Resource
    private MessageDecoder messageDecoder;

    @Override
    public void handle(DomainEventNotice notice, TransactionCallback callback) {
        log.debug("收到领域事件通知,{}", notice);

        // 获取消息id
        String messageId = EventNoticeUtils.getMessageId(notice);

        // 丢弃已经在本地消息表存在的领域事件通知
        boolean exists = messageReceivedHistoryRepository.existsById(messageId);

        if (exists) {
            log.debug("领域事件通知重复消费,直接丢弃,{}", notice);
            return;
        }
        EventNoticeUtils.getLocalUser(notice).ifPresent(userContextString -> {
            LocalUser localUser = messageDecoder.decode(userContextString, LocalUser.class);
            LocalUserHolder.setLocalUser(localUser);
        });
        // 开启本地事务
        TransactionStatus transactionStatus = platformTransactionManager.getTransaction(transactionDefinition);
        try {
            // 记录本地消息表
            storeMessage(messageId);
            // 执行具体通知方法
            doSubscribe(notice);
            // 提交事务
            platformTransactionManager.commit(transactionStatus);

            // 执行回调
            callback.invoke(notice, true);

        } catch (Exception rollbackException) {
            platformTransactionManager.rollback(transactionStatus);
            log.error("本地事务提交失败", rollbackException);
            callback.invoke(notice, false);
        }
    }

    private void doSubscribe(DomainEventNotice notice) throws ClassNotFoundException {
        String classType = EventNoticeUtils.getClassType(notice);
        String classVersion = EventNoticeUtils.getClassVersion(notice);
        // 反序列化
        try {
            Object payload = messageDecoder.decode(notice.getPayload(),
                    eventTypeLoader.loadTargetClass(classType, classVersion));
            applicationEventPublisher.publishEvent(payload);
        } catch (ClassNotFoundException e) {
            log.error("反序列化失败,类型{}无法加载", classType);
            throw e;
        }

    }

    private void storeMessage(String messageId) {
        MessageReceivedHistory history = MessageReceivedHistory.builder()
                .id(messageId)
                .createdDate(Instant.now())
                .build();

        messageReceivedHistoryRepository.save(history);
    }
}
